#pragma once

#include "my_disk_layout.h"
#include "comm/my_atomic.h"
#include "comm/my_rwlock.h"
#include "comm/my_spin_lock.h"
#include "comm/my_defs.h"
#include "event/my_aio_manager.h"

#define MY_DISK_VOLUME_MIN_DATA_SPACE_SIZE          (32 * MY_DISK_BLOCK_SIZE)
#define MY_DISK_VOLUME_MIN_RESERVE_FREE_SPACE_SIZE  (4 * MY_DISK_BLOCK_SIZE)
#define MY_DISK_VOLUME_MIN_RESERVE_USED_SPACE_SIZE  (4 * MY_DISK_BLOCK_SIZE)

/***
 * 基本原理:
 * 将data_space看做一个圆. 数据从recycle_pos开始到write_pos结束.
 * - recycle_pos到write_pos只是这个圆上的一段路径
 * - recycle_pos不可超过write_pos,
 * - write_pos亦不可反超recycle_pos
 * 
 * 为了保证recycle_pos与write_pos顺序, 固需要遵守如下约定
 * - 我们要求一个volume至少包含 MY_DISK_VOLUME_DATA_SPACE_SIZE 个BLOCK的空间
 * - 只有write_pos距离recycle_pos至少有4个BLOCK_SIZE时才可申请新的space以供写入.
 * - 只有recycle_pos距离write_pos至少有4个BLOCK_SIZE时才可申请回收.
 * - ps: 必须有2个BLOCK_SIZE的间隙才可保证上述流程安全运转 (由于滑动窗口的关系, 导致最后一部分空间不够一个BLOCK)
 * - 这里使用4个BLOCK_SIZE仅仅是为了满足上述的需求.
 */


typedef struct my_disk_vinfo_s my_disk_vinfo_t;

struct my_disk_vinfo_s {
    // 0: no close; 1: only close write; 2: close
    my_atomic_int_t                 close_stats;

    // static configuration
    int                             fd;             // file descriptor
    int64_t                         header_offset;  // volume header start offset
    int64_t                         index_offset;   // volume index start offset
    int64_t                         data_offset;    // data space start offset
    int64_t                         data_size;      // data space size

    // protect the following members
    my_rwlock_t                     rwlock;
    uint8_t                         write_serial;
    uint8_t                         recycle_serial;
    int64_t                         write_pos;
    int64_t                         recycle_pos;

    // protect the following members
    my_spin_lock_t                  spin_lock;
    my_disk_volume_header_t         *volume_header;
    // if header not modified, we skip write header in timer task
    my_bool_t                       header_modified;
};

static inline void
my_disk_vinfo_init(my_disk_vinfo_t *vinfo) {
    bzero(vinfo, sizeof(my_disk_vinfo_t));
    vinfo->fd = INVALID_FD;
    my_rwlock_init(&vinfo->rwlock);
    my_spin_lock_init(&vinfo->spin_lock);
}

static inline void
my_disk_vinfo_close(my_disk_vinfo_t *vinfo, my_bool_t only_close_write) {
    if (only_close_write) {
        my_atomic_int_swap(&vinfo->close_stats, 1);
        return;
    }

    my_atomic_int_swap(&vinfo->close_stats, 2);
}

static inline my_bool_t
my_disk_vinfo_closed_write(my_disk_vinfo_t *vinfo) {
    return my_atomic_int_load(&vinfo->close_stats) != 0;
}

static inline my_bool_t
my_disk_vinfo_closed(my_disk_vinfo_t *vinfo) {
    return my_atomic_int_load(&vinfo->close_stats) == 2;
}

static inline void  // unsafe
my_disk_vinfo_assert_vaild(my_disk_vinfo_t *vinfo) {
    assert(my_disk_serial_and_pos_vaild(vinfo->write_serial, vinfo->write_pos,
        vinfo->recycle_serial, vinfo->recycle_pos));
}

static inline void // unsafe
my_disk_vinfo_update_volume_header(my_disk_vinfo_t *vinfo) {
    my_disk_volume_header_t volume_header;
    my_disk_volume_header_init(&volume_header);
    volume_header.write_serial = vinfo->write_serial;
    volume_header.recycle_serial = vinfo->recycle_serial;
    volume_header.write_pos = vinfo->write_pos;
    volume_header.recycle_pos = vinfo->recycle_pos;

    if (!vinfo->volume_header) {
        MY_LOG_FATAL0("disk_vinfo", "volume not inited, the volume header never nil");
        return;
    }

    volume_header.index_item_num = vinfo->volume_header->index_item_num;
    my_disk_volume_header_update_checksum(&volume_header);

    my_spin_lock_lock(&vinfo->spin_lock);
    *(vinfo->volume_header) = volume_header;
    vinfo->header_modified = MY_TRUE;
    my_spin_lock_unlock(&vinfo->spin_lock);
}

/***
 * case1: recycle_serial == write_serial
 * |--invalid data--|recycle_pos--valid data--write_pos|--invalid data--|
 * 
 * case2: recycle_serial + 1 == write_serial
 * |--valid data--write_pos|--invalid data--|recycle_pos---valid data---|
 */
static inline int64_t // unsafe
my_disk_vinfo_free_space_size_helper(my_disk_vinfo_t *vinfo) {
    if (vinfo->recycle_serial + 1 == vinfo->write_serial) {
        return vinfo->recycle_pos - vinfo->write_pos;
    }

    // vinfo->recycle_serial == vinfo->write_serial
    return vinfo->data_size - vinfo->write_pos + vinfo->recycle_pos;
}

static inline int64_t
my_disk_vinfo_free_space_size(my_disk_vinfo_t *vinfo) {
    int64_t res = 0;
    my_rwlock_rlock(&vinfo->rwlock);
    res = my_disk_vinfo_free_space_size_helper(vinfo);
    my_rwlock_runlock(&vinfo->rwlock);
    return res;
}

/***
 * case1: recycle_serial == write_serial
 * |--invalid data--|recycle_pos--valid data--write_pos|--invalid data--|
 * 
 * case2: recycle_serial + 1 == write_serial
 * |--valid data--write_pos|--invalid data--|recycle_pos---valid data---|
 */
static inline int64_t // unsafe
my_disk_vinfo_used_space_size_helper(my_disk_vinfo_t *vinfo) {
    if (vinfo->recycle_serial + 1 == vinfo->write_serial) {
        return vinfo->data_size - vinfo->recycle_pos + vinfo->write_pos;
    }

    // recycle_serial == write_serial
    return vinfo->write_pos - vinfo->recycle_pos;
}

static inline int64_t
my_disk_vinfo_used_space_size(my_disk_vinfo_t *vinfo) {
    int64_t res = 0;
    my_rwlock_rlock(&vinfo->rwlock);
    res = my_disk_vinfo_used_space_size_helper(vinfo);
    my_rwlock_runlock(&vinfo->rwlock);
    return res;
}

/***
 * case1: recycle_serial == write_serial
 * |--invalid data--|recycle_pos--valid data--write_pos|--invalid data--|
 * 
 * case2: recycle_serial + 1 == write_serial
 * |--valid data--write_pos|--invalid data--|recycle_pos---valid data---|
 */
static inline my_bool_t // unsafe
my_disk_vinfo_vaild_index_item_helper(my_disk_vinfo_t *vinfo, 
    my_disk_index_item_t *item) {

    // invalid digest
    if (item->digest32 == 0) {
        return MY_FALSE;
    }

    // invalid serial number, maybe overwrite
    if (item->serial < vinfo->recycle_serial ||
        item->serial > vinfo->write_serial) {
        return MY_FALSE;
    }

    int64_t offset = my_disk_index_item_offset(item);
    // write_pos left and recycle_pos right are vaild
    if (vinfo->recycle_serial == vinfo->write_serial) {
        if (offset < vinfo->recycle_pos || offset > vinfo->write_pos) {
            return MY_FALSE;
        }
        return MY_TRUE;
    }

    // recycle_serial + 1 == write_serial && item->serial == recycle_serial
    // recycle_pos right are vaild
    if (item->serial == vinfo->recycle_serial) {
        if (offset < vinfo->recycle_pos) {
            return MY_FALSE;
        }
        return MY_TRUE;
    }

    // recycle_serial + 1 == write_serial && item->serial == write_serial
    // write_pos left are vaild
    if (offset > vinfo->write_pos) {
        return MY_FALSE;
    }

    return MY_TRUE;
}

static inline my_bool_t
my_disk_vinfo_vaild_index_item(my_disk_vinfo_t *vinfo, 
    my_disk_index_item_t *item) {

    my_bool_t res = MY_FALSE;
    my_rwlock_rlock(&vinfo->rwlock);
    res = my_disk_vinfo_vaild_index_item_helper(vinfo, item);
    my_rwlock_runlock(&vinfo->rwlock);
    return res;
}

typedef struct {
    uint8_t vaild;
    int64_t recycle_pos;
    int64_t recycle_len;
} my_disk_vinfo_recycle_cursor_t;

// this only gives the reclaimable space
static inline my_disk_vinfo_recycle_cursor_t
my_disk_vinfo_next_recycle_cursor(my_disk_vinfo_t *vinfo) {
    my_disk_vinfo_recycle_cursor_t recycle_cursor;
    recycle_cursor.vaild = 0;  // set invalid

    // my_rwlock_wlock(&vinfo->rwlock);
    // if (vinfo->recycle_pos >= vinfo->data_size) {
    //     vinfo->recycle_serial++;
    //     vinfo->recycle_pos = 0;
    // }
    // my_rwlock_wunlock(&vinfo->rwlock);

    static int64_t tow_block_size = MY_DISK_BLOCK_SIZE * 2;
    my_rwlock_rlock(&vinfo->rwlock);
    int64_t used_size = my_disk_vinfo_used_space_size(vinfo);
    if (used_size > MY_DISK_VOLUME_MIN_RESERVE_USED_SPACE_SIZE) {
        // recycling usually takes the block size as the unit
        // the final space may be less than block size.
        // so we will recycle end-1 and end block at the same time.
        recycle_cursor.vaild = 1;
        recycle_cursor.recycle_pos = vinfo->recycle_pos;
        recycle_cursor.recycle_len = MY_DISK_BLOCK_SIZE;
        if (vinfo->recycle_pos + tow_block_size > vinfo->data_size) {
            recycle_cursor.recycle_len = vinfo->data_size - vinfo->recycle_pos;
        }
    }
    my_rwlock_runlock(&vinfo->rwlock);

    return recycle_cursor;
}

// @params: recycle len is length of real recovery.
// 当recycle cursor中仅包含某个对象的部分时, 这表明该recycle_space已经无法回收更多的数据.
// 这个对象需要在下次recycle时进行回收.
// [--recycle_space--]
// |--data1--|--data2--|
// [---------] ----- real recycle len
//           [--recycle_space--]
//           [-----------------] -- next recycle
static inline void
my_disk_vinfo_recycle_done(my_disk_vinfo_t *vinfo, int64_t recycle_len) {
    if (!MY_IS_PAGE_ALIGNED(recycle_len)) {
        MY_LOG_FATAL("disk_vinfo", "unreachable, recycle len: %d", recycle_len);
    }

    my_rwlock_wlock(&vinfo->rwlock);
    if (vinfo->recycle_pos + recycle_len >= vinfo->data_size) {
        recycle_len = recycle_len - vinfo->data_size + vinfo->recycle_pos;
        vinfo->recycle_pos = 0;
        vinfo->recycle_serial++;
    }
    vinfo->recycle_pos += recycle_len;
    my_rwlock_wunlock(&vinfo->rwlock);

    my_rwlock_rlock(&vinfo->rwlock);
    my_disk_vinfo_assert_vaild(vinfo);
    my_disk_vinfo_update_volume_header(vinfo);
    my_rwlock_runlock(&vinfo->rwlock);
}

typedef struct {
    uint8_t     vaild;
    uint8_t     write_serial;
    int64_t     write_pos;
} my_disk_vinfo_write_cursor_t;

static inline my_disk_vinfo_write_cursor_t
my_disk_vinfo_next_write_cursor(my_disk_vinfo_t *vinfo) {
    my_disk_vinfo_write_cursor_t write_cursor;
    write_cursor.vaild = 0; // set invalid

    my_rwlock_wlock(&vinfo->rwlock);
    int64_t free_size = my_disk_vinfo_free_space_size_helper(vinfo);
    if (free_size > MY_DISK_VOLUME_MIN_RESERVE_FREE_SPACE_SIZE) {
        write_cursor.vaild = 1;
        // when the final space is less than a blocksize, it will be skipped.
        if (vinfo->write_pos + MY_DISK_BLOCK_SIZE > vinfo->data_size) {
            vinfo->write_serial++;
            vinfo->write_pos = 0;
        }
        write_cursor.write_serial = vinfo->write_serial;
        write_cursor.write_pos = vinfo->write_pos;
        // take up space for index item vaild check
        vinfo->write_pos += MY_DISK_BLOCK_SIZE;
    }
    my_rwlock_wunlock(&vinfo->rwlock);

    my_rwlock_rlock(&vinfo->rwlock);
    my_disk_vinfo_assert_vaild(vinfo);
    my_disk_vinfo_update_volume_header(vinfo);
    my_rwlock_runlock(&vinfo->rwlock);
    return write_cursor;
}

static inline void
my_disk_vinfo_write_done(my_disk_vinfo_t *vinfo, int64_t write_len) {
    if (!MY_IS_PAGE_ALIGNED(write_len)) {
        MY_LOG_FATAL("disk_vinfo", "unreachable, write len: %d", write_len);
    }

    // apply for a block of space in my_disk_vinfo_next_write_cursor
    // in the case of using part, we need to return the remaining
    int64_t back_size = MY_DISK_BLOCK_SIZE - write_len;
    if (back_size < 0) {
        MY_LOG_FATAL("disk_vinfo", "the data written at one time will not exceed block "
            "size, write_len: %d", write_len);
    }

    my_rwlock_wlock(&vinfo->rwlock);
    vinfo->write_pos -= back_size;
    my_rwlock_wunlock(&vinfo->rwlock);
}